package cn.ac.iie.ulss.dataredistribution.handler;

import cn.ac.iie.ulss.dataredistribution.commons.GlobalVariables;
import cn.ac.iie.ulss.dataredistribution.commons.RuntimeEnv;
import cn.ac.iie.ulss.dataredistribution.tools.HNode;
import cn.ac.iie.ulss.dataredistribution.consistenthashing.NodeLocator;
import cn.ac.iie.ulss.dataredistribution.tools.RData;
import cn.ac.iie.ulss.dataredistribution.tools.Rule;
import cn.ac.iie.ulss.dataredistribution.tools.INode;
import java.io.ByteArrayInputStream;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.log4j.PropertyConfigurator;
import org.apache.avro.util.Utf8;

/**
 *
 * @author: evan
 * @date: 2014-10-15
 */
public class TransmitThread implements Runnable {

    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss");
    String topic = null;
    ConcurrentLinkedQueue dataPool = null;
    String docSchemaContent = null;
    Protocol docProtocol = null;
    Schema docSchema = null;
    DatumReader<GenericRecord> docReader = null;
    int uselessfile = 0;
    int sendSize = 1000;
    ConcurrentHashMap<String, String> keyWordToType = null;
    ConcurrentHashMap<String, Rule> tableToRule = null;
    ConcurrentHashMap<String, ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>>> tableToMSGStation = null;
    ConcurrentHashMap<String, INode> intervalToSNode = null;
    static org.apache.log4j.Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = org.apache.log4j.Logger.getLogger(TransmitThread.class.getName());
    }

    public TransmitThread(ConcurrentLinkedQueue dataPool, String topic) {
        this.dataPool = dataPool;
        this.topic = topic;
    }

    @Override
    public void run() {
        init();
        logger.info("begining the dataSplit and send the message from " + topic + " to the transfer station ");
        while (true) {
            if (!dataPool.isEmpty()) {
                try {
                    dataSplitAndSend();
                } catch (Exception ex) {
                    logger.error("a transmit thread for " + topic + " is dead " + ex, ex);
                    return;
                }
            } else {
                logger.debug("dataPool for the topic " + topic + " is empty !");
                try {
                    Thread.sleep(200);
                } catch (Exception ex) {
                    //logger.info(ex, ex);
                }
            }
        }
    }

    /**
     *
     * init the environment
     */
    private void init() {
        docSchemaContent = (String) RuntimeEnv.getParam(GlobalVariables.DOC_SCHEMA_CONTENT);
        docProtocol = Protocol.parse(docSchemaContent);
        uselessfile = (Integer) RuntimeEnv.getParam(RuntimeEnv.USELESS_FILE);
        sendSize = (Integer) RuntimeEnv.getParam(RuntimeEnv.SEND_SIZE);
        tableToRule = (ConcurrentHashMap<String, Rule>) RuntimeEnv.getParam(GlobalVariables.TABLE_TO_RULE);
        keyWordToType = (ConcurrentHashMap<String, String>) RuntimeEnv.getParam(GlobalVariables.KEYWORD_TO_TYPE);
        tableToMSGStation = (ConcurrentHashMap<String, ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>>>) RuntimeEnv.getParam(GlobalVariables.TABLE_TO_MSGSTATION);
        intervalToSNode = (ConcurrentHashMap<String, INode>) RuntimeEnv.getParam(GlobalVariables.INTERVAL_TO_SNODE);
    }

    /**
     *
     * Split and send the message to the transfer station
     */
    public void dataSplitAndSend() {
        logger.debug("begining the dataSplit  message from " + topic + " and send to the transfer station ");
        RData data = null;
        GenericRecord msgRecord = null;
        ByteArrayInputStream msgbis = null;
        BinaryDecoder msgbd = null;
        while ((data = (RData) dataPool.poll()) != null) {
            docSchema = docProtocol.getType(GlobalVariables.DOC);
            docReader = new GenericDatumReader<GenericRecord>(docSchema);
            msgbis = new ByteArrayInputStream(data.getData());
            msgbd = new DecoderFactory().binaryDecoder(msgbis, null);
            try {
                msgRecord = docReader.read(null, msgbd);
            } catch (Exception ex) {
                logger.debug("split the one data from the topic " + topic + " in the dataPool wrong " + ex, ex);
                if (uselessfile == 1) {
                    storeUselessData(topic, data.getData());
                }
                continue;
            }
            sendToType(data, msgRecord, sendSize * 2);
        }
    }

    private void sendToType(RData data, GenericRecord record, int maxsize) {
        Rule rule = tableToRule.get(data.getTable());
        String[] keywords = rule.getKeywords().split("\\|");
        StringBuilder sb = new StringBuilder();
        for (int i = 1; i < keywords.length; i++) {
            String ss = keywords[i];
            
            String type = keyWordToType.get(data.getTable() + ss);
            
            Map<Utf8, Object> m = (Map<Utf8, Object>) record.get(type);
            
            if (m == null) {
                sb.append("");
            } else {
                String content = m.get(new Utf8(ss)).toString();
                if (content == null) {
                    sb.append("");
                } else {
                    sb.append(content);
                }
            }
        }
        NodeLocator n4 = rule.getNodelocator();
        HNode node = n4.getPrimary(sb.toString());
        Long time = 0L;
        String keyinterval = null;
        String keytime = keywords[0].toLowerCase();
        String type = keyWordToType.get(data.getTable() + keytime);       
        Map<Utf8, Long> m2 = (Map<Utf8, Long>) record.get(type);       
        time = m2.get(new Utf8(keytime));
        String unit = (rule.getPartType().split("\\|"))[3];
        String interval = (rule.getPartType().split("\\|"))[4];
        keyinterval = getKeyInterval(time, unit, interval);
        ConcurrentHashMap<INode, ConcurrentLinkedQueue> chm = tableToMSGStation.get(data.getTable()).get(node);
        synchronized (chm) {
            INode mn = intervalToSNode.get(data.getTable() + keyinterval);
            if (mn != null) {
                ConcurrentLinkedQueue clq = (ConcurrentLinkedQueue) chm.get(mn);
                if (clq != null) {
                    while (clq.size() > maxsize) {
                        logger.info("the clq for " + data.getTable() + " " + keyinterval + " " + node.getName() + " is full");
                        try {
                            Thread.sleep(200);
                        } catch (Exception ex) {
                            //do nothing
                        }
                    }
                    clq.offer(data.getData());
                } else {
                    logger.info("create a abq for " + data.getTable() + " " + keyinterval + " " + node.getName());
                    clq = new ConcurrentLinkedQueue();
                    clq.offer(data.getData());
                    chm.put(mn, clq);
                }
            } else {
                ConcurrentLinkedQueue clq = new ConcurrentLinkedQueue();
                clq.offer(data.getData());
                mn = new INode(keyinterval, System.currentTimeMillis());
                chm.put(mn, clq);
                intervalToSNode.put(data.getTable() + keyinterval, mn);
            }
        }
    }
    
    /**
     *
     * place the useless data to the uselessDataStore
     */
    private void storeUselessData(String topic, byte[] data) {
        ConcurrentHashMap<String, ConcurrentLinkedQueue> uselessDataStore = (ConcurrentHashMap<String, ConcurrentLinkedQueue>) RuntimeEnv.getParam(GlobalVariables.USELESS_DATA_STORE);
        synchronized (RuntimeEnv.getParam(GlobalVariables.SYN_STORE_USELESSDATA)) {
            if (uselessDataStore.containsKey(topic)) {
                ConcurrentLinkedQueue clq = uselessDataStore.get(topic);
                clq.offer(data);
            } else {
                ConcurrentLinkedQueue sdQueue = new ConcurrentLinkedQueue();
                sdQueue.offer(data);
                uselessDataStore.put(topic, sdQueue);
                StoreUselessDataThread sudt = new StoreUselessDataThread(sdQueue, topic);
                Thread tsudt = new Thread(sudt);
                tsudt.setName("StoreUselessDataThread-" + topic);
                tsudt.start();
                logger.info("start a StoreUselessDataThread for " + topic);
            }
        }
    }

    /**
     *
     * get the keyinterval by time ,unit and interval
     */
    private String getKeyInterval(Long time, String unit, String interval) {
        String st = null;
        String et = null;

        Calendar cal = Calendar.getInstance();
        cal.setTimeInMillis(time * 1000);
        String keyinterval = null;

        if ("'MI'".equalsIgnoreCase(unit)) {   //以分钟为单位
            cal.set(14, 0);
            cal.set(13, 0);
            cal.set(12, cal.get(12) - cal.get(12) % Integer.parseInt(interval));
            st = dateFormat.format(cal.getTime());
            if (cal.get(12) + Integer.parseInt(interval) > 60) {
                cal.set(12, 60);
            } else {
                cal.set(12, cal.get(12) + Integer.parseInt(interval));
            }
            et = dateFormat.format(cal.getTime());
            keyinterval = st + "|" + et;
        } else if ("'H'".equalsIgnoreCase(unit)) {    //以小时为单位
            cal.set(14, 0);
            cal.set(13, 0);
            cal.set(12, 0);
            cal.set(11, cal.get(11) - cal.get(11) % Integer.parseInt(interval));
            st = dateFormat.format(cal.getTime());
            if (cal.get(11) + Integer.parseInt(interval) > 24) {
                cal.set(11, 24);
            } else {
                cal.set(11, cal.get(11) + Integer.parseInt(interval));
            }
            et = dateFormat.format(cal.getTime());
            keyinterval = st + "|" + et;
        } else if ("'D'".equalsIgnoreCase(unit)) { //以天为单位
            cal.set(14, 0);
            cal.set(13, 0);
            cal.set(12, 0);
            cal.set(11, 0);
            cal.set(6, cal.get(6) - cal.get(6) % Integer.parseInt(interval));
            st = dateFormat.format(cal.getTime());
            cal.set(6, cal.get(6) + Integer.parseInt(interval));
            et = dateFormat.format(cal.getTime());
            keyinterval = st + "|" + et;
        } else {
            logger.error("now the partition unit is not support, it only supports --- D day,H hour,MI minute");
        }
        return keyinterval;
    }
}